package com.kafka.producer;

import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import net.sf.json.JSONObject;

import com.kafka.bean.Device;
import com.kafka.common.Constant;
import com.kafka.common.String.StringValues;
import com.kafka.core.BaseProducer;
import com.kafka.dao.DeviceMapper;
import com.kafka.dao.UserMapper;
import com.kafka.enumtype.MsgType;
import com.kafka.producer.inter.IProducer;
import com.kafka.util.BaseUtil;
import com.kafka.valid.DeviceValid;

/**
 * 注册消息类
 * 
 * @author leihongan
 * 
 */
public class DeviceProducer extends BaseProducer implements IProducer {
	UserMapper userDao = sqlSession.getMapper(UserMapper.class);
	DeviceMapper deviceDao = sqlSession.getMapper(DeviceMapper.class);
	private final Producer<String, String> producer;
	public String TOPIC;

	public DeviceProducer() {
		Properties properties = new Properties();
		try {
			properties.load(ClassLoader
					.getSystemResourceAsStream("device.properties"));
			TOPIC = properties.getProperty("topic");
		} catch (IOException e) {
			e.printStackTrace();
		}
		producer = new Producer<String, String>(new ProducerConfig(properties));
	}

	public String produce(JSONObject message) {
		if (MsgType.BINDDEVICE.toString().equals(message.getString("msgtype"))){
			return bindProduce(message);
		} else if (MsgType.UNBINDDEVICE.toString().equals(message.getString("msgtype"))) {
			return unbindProduce(message);
		} else {
			return null;
		}
	}
	
	public String bindProduce(JSONObject message) {
		try {
			String msgId = message.getString("msgid");
			// 校验数据合法性
			JSONObject ck = DeviceValid.validBind(rebuildJSON(message.getJSONObject("datas")), msgId);
			if (ck != null) {
					addMessage(message,TOPIC,Constant.FAIL, ck.getString("msg"));
					// 如果错误信息不为空，返回错误信息
					return ck.toString();
			} else {
				// 追加appid和secretkey
				addAppidAndSecretkey(message);
				Map<String, Object> params = new HashMap<>();
				params.put("status", Constant.SUCCESS);
				params.put("message_id", message.get("msgid"));
				if(messageDao.selectByParam(params)==0){
					Map<String,String> query = new HashMap<>();
					query.put("dcode", message.getJSONObject("datas").getString("dcode"));
					Device device = deviceDao.selectByCondition(query);
					// 如果存在对应的uuid并通过校验则绑定腕表
					// 判断腕表是否之前已经绑定过
					if(device==null){
						// 发送消息keyedMessage第一个参数是topic第二个参数就是发送的消息内容
						producer.send(new KeyedMessage<String, String>(TOPIC, message.toString()));
						// 未绑定过新增
						addDevice(message);
					}else{
						// 已绑定过更新状态 且状态为解绑
						if(device.getEnabled() != Constant.DEVICEACTIVATE){
							// 发送消息keyedMessage第一个参数是topic第二个参数就是发送的消息内容
							producer.send(new KeyedMessage<String, String>(TOPIC, message.toString()));
							device.setEnabled(Constant.DEVICEACTIVATE);
							deviceDao.updateByPrimaryKey(device);
						}else{
							return failMsg(message.get("msgid"),StringValues.DEVICE_HAVE_BEEN_BIND);
						}
					}
					log.info("###DeviceProducer:" + message.toString() + ",time:" + BaseUtil.getDate());
					addMessage(message,TOPIC,Constant.SUCCESS, StringValues.SEND_SUCCESSFUL);
					return successMsg(message.get("msgid"));
				}else{
					return failMsg(message.get("msgid"),StringValues.SEND_FAIL);
				}
			}
		} catch (Exception e) {
			addMessage(message,TOPIC,Constant.FAIL, e.toString());
			return failMsg(message.get("msgid"), e.toString());
		}
	}
	
	public String unbindProduce(JSONObject message) {
		try {
			String msgId = message.getString("msgid");
			// 校验数据合法性
			JSONObject ck = DeviceValid.validUnbind(rebuildJSON(message.getJSONObject("datas")), msgId);
			if (ck != null) {
					addMessage(message,TOPIC,Constant.FAIL, ck.getString("msg"));
					// 如果错误信息不为空，返回错误信息
					return ck.toString();
			} else {
				// 追加appid和secretkey
				addAppidAndSecretkey(message);
				Map<String, Object> params = new HashMap<>();
				params.put("status", Constant.SUCCESS);
				params.put("message_id", message.get("msgid"));
				if(messageDao.selectByParam(params)==0){
					Map<String,String> query = new HashMap<>();
					query.put("dcode", message.getJSONObject("datas").getString("dcode"));
					Device device = deviceDao.selectByCondition(query);
					if(device.getEnabled()==Constant.DEVICEUNBIND){
						return failMsg(message.get("msgid"),StringValues.DEVICE_HAVE_BEEN_UNBIND);
					}else{
						// 发送消息keyedMessage第一个参数是topic第二个参数就是发送的消息内容
						producer.send(new KeyedMessage<String, String>(TOPIC, message.toString()));
						// 如果存在对应的uuid并通过校验则解绑腕表
						updateDevice(message, Constant.DEVICEUNBIND);
					}
					log.info("###DeviceProducer:" + message.toString() + ",time:" + BaseUtil.getDate());
					addMessage(message,TOPIC,Constant.SUCCESS, StringValues.SEND_SUCCESSFUL);
					return successMsg(message.get("msgid"));
				}else{
					return failMsg(message.get("msgid"),StringValues.SEND_FAIL);
				}
			}
		} catch (Exception e) {
			addMessage(message,TOPIC,Constant.FAIL, e.toString());
			return failMsg(message.get("msgid"), e.toString());
		}
	}

	/**
	 * json数据处理方法
	 * 
	 * @param message
	 * @return
	 */
	private void addDevice(JSONObject message) throws Exception {
		JSONObject deviceData = message.getJSONObject("datas");
		Device device = createEntity(deviceData);
		deviceDao.insert(device);
		sqlSession.commit();
	}
	
	private void updateDevice(JSONObject message, Integer status){
		JSONObject deviceData = message.getJSONObject("datas");
		deviceDao.updateByUUIDAndDCode(String.valueOf(status),deviceData.getString("uuid"), deviceData.getString("dcode"));
		sqlSession.commit();
	}

	/**
	 * 创建Device实体
	 * 
	 * @param accountData
	 * @return
	 * @throws ParseException
	 */
	private Device createEntity(JSONObject deviceData) throws Exception {
		Device device = new Device();
		device.setUuid(deviceData.getString("uuid"));
		device.setDtype(Integer.valueOf(deviceData.getString("dtype")));
		device.setDname(deviceData.getString("dname"));
		device.setDcode(deviceData.getString("dcode"));
		device.setModel(deviceData.getString("model"));
		device.setManu(deviceData.getString("manu"));
		device.setEnabled(Constant.DEVICEACTIVATE);
		SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd");
		Date date = format.parse(deviceData.getString("producedt"));
		device.setProducedt(date);
		SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
		Date createTime = sdf.parse(BaseUtil.getDate());
		device.setCreateTime(createTime);
		return device;
	}
}